package com.alibaba.otter.node.etl.load.loader.mq.rabbit;

import com.alibaba.otter.node.etl.load.loader.mq.rabbit.assembly.RabbitMqMessageAssembler;
import com.alibaba.otter.node.etl.load.loader.mq.rabbit.callback.ConnectedCallback;
import com.alibaba.otter.node.etl.load.loader.mq.rabbit.callback.DisconnectedCallback;
import com.alibaba.otter.node.etl.load.loader.mq.rabbit.callback.PublishFailureCallback;
import com.alibaba.otter.node.etl.load.loader.mq.rabbit.callback.SuccessCallback;
import com.alibaba.otter.node.etl.load.loader.mq.rabbit.model.MessageType;
import com.alibaba.otter.node.etl.load.loader.mq.rabbit.provider.DefaultContentTypeProvider;
import com.alibaba.otter.node.etl.load.loader.mq.rabbit.provider.HeaderProvider;
import com.alibaba.otter.node.etl.load.loader.mq.rabbit.provider.PropertyProvider;
import org.apache.commons.lang3.ArrayUtils;

import java.util.Arrays;

/**
 * Publisher构造器
 *
 * @author dengchao
 */

public class PublisherBuilder {

    private RabbitMqMessageAssembler messageAssembler = new RabbitMqMessageAssembler();
    private RabbitMqPublisher publisher;

    PublisherBuilder(String namespaceName) {
        this.publisher = new RabbitMqPublisher(namespaceName, this.messageAssembler);
    }

    public Publisher build(MessageType type) {
        this.messageAssembler.setSerializer(type.newSerializer());
        this.messageAssembler.addPropertyProviders(
            Arrays.asList(new DefaultContentTypeProvider(type)));
        this.messageAssembler.setExchange(publisher.getExchange());
        this.messageAssembler.setRoutingKey(publisher.getRoutingKey());
        return publisher;
    }

    public Publisher build() {
        return build(MessageType.JAVA_TYPE);
    }

    public PublisherBuilder toExchange(String exchange) {
        publisher.setExchange(exchange);
        return this;
    }

    public PublisherBuilder routingKey(String routingKey) {
        publisher.setRoutingKey(routingKey);
        return this;
    }

    public PublisherBuilder onFailure(PublishFailureCallback callback) {
        publisher.setFailureCallback(callback);
        return this;
    }

    public PublisherBuilder onSuccess(SuccessCallback... successCallbacks) {
        if (ArrayUtils.isNotEmpty(successCallbacks)) {
            publisher.addSuccessCallbacks(Arrays.asList(successCallbacks));
        }
        return this;
    }

    public PublisherBuilder onConnected(ConnectedCallback callback) {
        publisher.setConnectedCallback(callback);
        return this;
    }

    public PublisherBuilder onDisconnected(DisconnectedCallback callback) {
        publisher.setDisconnectedCallback(callback);
        return this;
    }

    public PublisherBuilder customHeader(HeaderProvider... headerProviders) {
        if (headerProviders != null) {
            this.messageAssembler.addHeaderProviders(Arrays.asList(headerProviders));
        }
        return this;
    }

    public PublisherBuilder customProperty(PropertyProvider... propertyProviders) {
        if (propertyProviders != null) {
            this.messageAssembler.addPropertyProviders(Arrays.asList(propertyProviders));
        }
        return this;
    }

    public PublisherBuilder requireConfirm(boolean requireConfirm) {
        publisher.setRequireConfirm(requireConfirm);
        return this;
    }
}
